package drpc;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.StormSubmitter;
import org.apache.storm.drpc.LinearDRPCTopologyBuilder;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 简单例子，实现DRPC
 */
public class BasicDRPCTopology {
    private static final Logger logger = LoggerFactory.getLogger(BasicDRPCTopology.class);
    public static void main(String[] args) throws Exception {
        String topologyName = "drpc_demo";
        String function = "exclamation";

        LinearDRPCTopologyBuilder builder = new LinearDRPCTopologyBuilder(function);
        // 绑定bolt组件
        builder.addBolt(new ExclaimBolt(), 3);

        Config config = new Config();

        if (args == null || args.length == 0) {
            LocalDRPC localDRPC = new LocalDRPC();
            LocalCluster localCluster = new LocalCluster();

            localCluster.submitTopology(topologyName, config, builder.createLocalTopology(localDRPC));

            for (String arg : new String[]{"a", "b", "c", "d"}) {
                String result = localDRPC.execute(function, arg);
                logger.info("Result for {}: {}", arg, result);
            }
        } else {
            StormSubmitter.submitTopologyWithProgressBar(topologyName, config, builder.createRemoteTopology());
        }
    }

    public static class ExclaimBolt extends BaseBasicBolt {

        @Override
        public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
            String args = tuple.getString(1);// 从0开始
            basicOutputCollector.emit(new Values(tuple.getValue(0), args + "!"));
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
            outputFieldsDeclarer.declare(new Fields("id", "result"));
        }
    }
}
